package com.teemor.common.bean.utils.kafka;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.bouncycastle.voms.VOMSAttribute;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/2/6
 */
@Component
@Slf4j
public class BaseKafkaProducer {

    private static KafkaTemplate<Integer, String> kafkaTemplate;

    public static void setKafkaTemplate(KafkaTemplate<Integer, String> template) {
        if (kafkaTemplate == null) {
            synchronized (String.class) {
                if (kafkaTemplate == null) {
                    kafkaTemplate = template;
                }
            }
        }
    }

    public static void sendMessage(ProducerRecord<Integer, String> record) {
        ListenableFuture<SendResult<Integer, String>> result = kafkaTemplate.send(record);
        result.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("消息发送失败", throwable);
                System.out.println("发送失败");
            }

            @Override
            public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
                log.debug("消息发送成功");
            }
        });
    }

}
